<?php

namespace Server\Queue;

use lactone\Core\network\ClientInstance;
use lactone\Core\Queue\Queue;
use lactone\Core\Models\TaskModel;
use lactone\Core\Spider\Spiders;

class TaskSendQueue extends Queue{

    public $ShiftLoopTime = 500;      //自动读取时间间隔(毫秒)

    public $pushLoopTime = 600;       //自动写入读取时间间隔(毫秒)

    public $Driver = "Array";

    public $max    = 20;            //队列中最大的数据总量

    public $limit = 5;              //数据查询sql limit

    private $is_first = true;

    protected $Clients = [];

    function init(){
        if(LACTONE_DEBUG){
            $this->loadDriver();

            $this->lock  = new \swoole_lock();
        }else{
            parent::init();
        }
    }

    public function run(){
        if(LACTONE_DEBUG){
            $this->autoPush();
        }
    }

    public function _do($data){

        try{
            if($data instanceof TaskModel){

                \Lactone::info("do taskModel:".$data->id,'doTaskSendQueue');


                if(!isset($this->Clients[$data->s_id])){
                    $Spider = \Lactone::$Spiders->findById($data->s_id);
                    $ClientInstance = new ClientInstance(['ip'=>$Spider->ip,'port'=>$Spider->port]);
                    if($ClientInstance->initClient(false,SWOOLE_SOCK_TCP | SWOOLE_KEEP)){
                        $this->Clients[$data->s_id] = $ClientInstance;
                    }else{
                        $data->update_attribute("status",TaskModel::Wait_Next_Status);
                        $Spider->update_attribute("status",Spiders::Dead_Status);
                        \Lactone::$Spiders->stepCutTask($Spider->id);

                        \Lactone::warning("[ip:$Spider->ip,port:$Spider->port] Connect Faild! ",'doTaskSendQueue');

                        return false;
                    }
                }

                //发送失败任务退回到等待状态
                if(!$this->Clients[$data->s_id]->send($data->__toString())){
                    $Spider = \Lactone::$Spiders->findById($data->s_id);
                    \Lactone::warning("[ip:$Spider->ip,port:$Spider->port] Send Faild! ",'doTaskSendQueue');

                    $data->update_attribute("status",TaskModel::Wait_Next_Status);
                    $Spider->update_attribute("status",Spiders::Dead_Status);
                    \Lactone::$Spiders->stepCutTask($Spider->id);

                    unset($this->Clients[$data->s_id]);
                }else{
                    $data->update_attribute("status",TaskModel::Working_Status);

                }

                unset($data);
            }
        }catch (\Exception $e){
           \Lactone::log_Exception($e);
        }

    }


    /**
     * 自动push
     * @return bool
     * @throws \ActiveRecord\RecordNotFound
     */
    public function autoPush(){

        if(rand(1,50) ==1){
            \Lactone::info($this->logMessage('autoPush开始',__METHOD__));
        }

        try{

            //系统第一次运行
            if($this->isFirstRun()){
                $this->createFirstTask();
                return;
            }

            //没有活着的蜘蛛
            if(!\Lactone::$Spiders->hasLiveSpider()){
                \Lactone::info($this->logMessage("没有活着的蜘蛛!"));
                return;
            }

            //系统中蜘蛛繁忙，暂时停止
            if($Spider = \Lactone::$Spiders->getFreeSpider()){
                if( ($Spider->task_run - $Spider->task_run_done) > $this->max ){
                    \Lactone::info($this->logMessage("大家都很忙没有空闲!"));
                    return;
                }
            }


            //这里开启事务
            $Queue = $this;
            $transaction = function() use ($Queue) {
                $rand = rand(1,100);
                //从来没有执行过的数据
                if($rand>=1 && $rand <= 60){
                    $Tasks = TaskModel::find_by_sql(" select * from task where status =  ".TaskModel::Wait_Status." order by level desc limit ".$Queue->limit." for update ");
                }else if($rand > 60 && $rand<=80 ){
                    $time = time()-60;
                    $Tasks = TaskModel::find_by_sql(" select * from task where status =  ".TaskModel::Fail_Status." and update_at < ".$time." order by level desc,id ASC limit ".$Queue->limit."  for update ");
                }else{
                    $Tasks = TaskModel::find_by_sql(" select * from task where status =  ".TaskModel::Wait_Next_Status."  order by level desc limit  ".$Queue->limit." for update ");
                }


                if($Tasks){
                    foreach($Tasks as $task){
                        if($Spider = \Lactone::$Spiders->getFreeSpider()){
                            $task->readonly(false);

                            \Lactone::debug($this->logMessage("push taskModel:".$task->id,'push'));

                            $task->update_attributes([
                                'status' => TaskModel::Working_Status,
                                's_id'   => $Spider->id
                            ]);

                            \Lactone::$Spiders->stepAddTask($Spider->id);

                            $Queue->push($task);
                        }
                    }
                }

                return true;
            };

            TaskModel::transaction($transaction);

        }catch (\Exception $e){
            \Lactone::log_Exception($e);
        }

        if(rand(1,200) ==1){
            \Lactone::info($this->logMessage('end auto push',__METHOD__));
        }

        return false;
    }


    /**
     * 系统第一次运行
     * @return bool
     */
    protected function isFirstRun(){
        if($this->is_first && TaskModel::is_empty()){
            return true;
        }
        return false;
    }

    /**
     * 创建第一个任务
     */
    protected function createFirstTask(){
        $sites = \Lactone::getConfig('site');
        foreach($sites as $site){
            $request = ['method'=>'get'];
            TaskModel::create(['level'=>50,'hash'=>TaskModel::create_hash($site,$request),'url'=>$site,'status'=>0,'request'=>serialize($request)]);
        }
    }



}
